package dk.almbrand.microservice.kafka.consumer.thread;

import dk.almbrand.microservice.exception.SystemException;

import dk.almbrand.microservice.kafka.consumer.ConsumerDaemon;
import dk.almbrand.microservice.kafka.consumer.ConsumerDaemonImpl;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;

public class ConsumerThreadControllerImpl implements ConsumerThreadController {
    ExecutorService executor;
    FutureTask<ConsumerDaemon> task;

    public ConsumerThreadControllerImpl() {

    }

    public void start(ConsumerDaemon consumer) throws SystemException {
        try {
            // Initiates a single thread
            executor = Executors.newSingleThreadExecutor();
            task = (FutureTask<ConsumerDaemon>)executor.submit((ConsumerDaemonImpl)consumer);
            executor.shutdown();
        } catch (Exception e) {
            throw new SystemException("Exception during consumer task", e);
        }
    }

    public void stop() {
        if (task != null) {
            task.cancel(true);
        }

        if (executor != null) {
            executor.shutdown();
        }
    }
}
